package com.itclass.queue;

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.Date;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.LockSupport;

/**
 * 抽象的数组优先队列
 *
 * @param <E>
 */
public abstract class AbstractPriorityArrayBlockingQueue<E> extends AbstractBlockingQueue<E> {

    /**
     * 默认队列大小
     */
    protected static final int DEFAULT_QUEUE_SIZE = 16;

    /**
     * 容量
     */
    protected final int capacity;

    /**
     * 队列
     */
    protected final Object[] queue;

    /**
     * 下节点柄句
     */
    protected final static VarHandle NODE_NEXT_HANDLE;
    /**
     * 上节点柄句
     */
    protected final static VarHandle NODE_PREV_HANDLE;

    /**
     *
     */
    protected final static VarHandle NODE_CONDITION_HANDLE;


    static {
        try {
            MethodHandles.Lookup lookup = MethodHandles.lookup();
            NODE_CONDITION_HANDLE = lookup.findVarHandle(AbstractPriorityArrayBlockingQueue.CasConditionObject.class,
                    "waiterNodeNext", AbstractPriorityArrayBlockingQueue.ConditionNode.class);
            NODE_NEXT_HANDLE = lookup.findVarHandle(Node.class, "next", Node.class);
            NODE_PREV_HANDLE = lookup.findVarHandle(Node.class, "prev", Node.class);
        } catch (NoSuchFieldException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    public AbstractPriorityArrayBlockingQueue() {
        this(DEFAULT_QUEUE_SIZE, Integer.MAX_VALUE);
    }

    public AbstractPriorityArrayBlockingQueue(int queueSize) {
        this(queueSize, Integer.MAX_VALUE);
    }

    public AbstractPriorityArrayBlockingQueue(int queueSize, int capacity) {
        this.queue = new Object[queueSize];
        this.capacity = capacity;
    }



    public int getQueueSize() {
        return queue.length;
    }

    /**
     * 获取优先级
     *
     * @param e 任务
     * @return
     */
    protected abstract int getPriority(E e);


    @Override
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        if (size() == capacity)
            return false;

        return doOffer(e);
    }

    @Override
    public E take() throws InterruptedException {


        return doTake();
    }

    @Override
    public E poll() {
        if (size() <= 0)
            return null;

        try {
            return doPoll(false);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

    }


    static class Node<E> {
        /**
         * 插入元素
         */
       protected   volatile  E item;
        /**
         * 下一个
         */
        protected volatile Node<E> next;

        /**
         * 上一个
         */
        protected  volatile Node<E> prev;

        Node(E x) {
            item = x;
        }
    }

    public static interface PriorityTask extends Runnable {
        /**
         * 获取任务的优先级，数值越大，优先级越高
         */
        int getPriority();
    }

    /**
     * cas 条件等待队列
     */
      class CasConditionObject implements Condition, java.io.Serializable {

        // 链表头节点。是单向链表
        private transient ConditionNode waiterNodeNext;
        @Override
        public void await() throws InterruptedException {
            // 获取线程中断标识
            if (Thread.interrupted())
                throw new InterruptedException();

            // 创建条件等待节点
            ConditionNode node = new ConditionNode();
            node.waiter = Thread.currentThread();

            // 往当前线程设置 parkBlocker 属性，属性值是当前对象
            LockSupport.setCurrentBlocker(this);
            boolean interrupted = false, cancelled = false, rejected = false;
            // 等于 0 线程需要等待
            while (size()==0) {
                // 判断线程是否要 中断
                if (interrupted |= Thread.interrupted()) {
                    // 设置当前线程为 放弃抢锁
                        break;
                }
                ConditionNode temp= waiterNodeNext;
                node.nextWaiter =temp;
                if (NODE_CONDITION_HANDLE.compareAndSet(this,temp,node)) {
                    try {
                        if (rejected)
                            // 线程等待
                            node.block();
                        else
                            // 线程等待
                            ForkJoinPool.managedBlock(node);
                    } catch (RejectedExecutionException ex) {
                        // 如果执行到这证明当前线程出现不能进入等待状态
                        rejected = true;
                    } catch (InterruptedException ie) {
                        // 出现这个错误证明该线程需要中断
                        interrupted = true;
                    }
                } else {
                    // 自旋等待下
                    Thread.onSpinWait();    // awoke while enqueuing

                }
            }
            /**
             * 代码执行到这证明该条件等待节点被唤醒了
             */
            LockSupport.setCurrentBlocker(null);
        }



        @Override
        public void awaitUninterruptibly() {

        }

        @Override
        public long awaitNanos(long nanosTimeout) throws InterruptedException {
            return 0;
        }

        @Override
        public boolean await(long time, TimeUnit unit) throws InterruptedException {
            return false;
        }

        @Override
        public boolean awaitUntil(Date deadline) throws InterruptedException {
            return false;
        }

        @Override
        public void signal() {
            ConditionNode first = waiterNodeNext;
            if (first != null)
                doSignal(first,false);
        }

        @Override
        public void signalAll() {
            ConditionNode first = waiterNodeNext;
            if (first != null)
                doSignal(first, true);
        }
        private void doSignal(ConditionNode first, boolean all) {
           do {
                if (NODE_CONDITION_HANDLE.compareAndSet(this,first,first.nextWaiter)){
                    ConditionNode next = first.nextWaiter;
                    enqueue(first);
                    if (all) {
                        break;
                    }
                    first = next;
                }else {
                    Thread.onSpinWait();
                    first = waiterNodeNext;
                }
            } while (first  != null);
        }
    }




    final void enqueue(ConditionNode node) {
        if (node != null) {
            LockSupport.unpark(node.waiter);
        }
    }
    class ConditionNode implements ForkJoinPool.ManagedBlocker {

        // 要等待的线程
        Thread waiter;
        // 下一个节点
        ConditionNode nextWaiter;            // link to next waiting node

        /**
         * Allows Conditions to be used in ForkJoinPools without
         * risking fixed pool exhaustion. This is usable only for
         * untimed Condition waits, not timed versions.
         */
        public final boolean isReleasable() {

            return size() > 0 || Thread.currentThread().isInterrupted();
        }

        public final boolean block() {

            // 如果队列元素>0的 就不需要去线程等待
            while (!isReleasable())
                LockSupport.park();
            return true;
        }
    }
}
